package io.task.cc.network.queue_task;

import io.task.cc.network.*;
import io.task.cc.network.utils.ByteBufferUtils;
import io.task.cc.network.utils.thread.pool.AbstractQueueTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;

/**
 * 解码任务对象，一个连接对应一个本对象
 *
 * @author huihui
 */
@Slf4j
@Setter
@Getter
public class DecodeQueueTask extends AbstractQueueTask<ByteBuffer> {

    private ChannelContext channelContext = null;
    private Config config = null;
    /**
     * 上一次解码剩下的数据
     */
    private ByteBuffer lastByteBuffer = null;
    /**
     * 新收到的数据
     */
    private ByteBuffer newReceivedByteBuffer = null;


    public DecodeQueueTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        this.config = channelContext.config;
        getQueue();
    }


    @Override
    public boolean addMsg(ByteBuffer packet) {
        return false;
    }

    public void handle(Packet packet, int byteCount) {
        channelContext.handleQueueRunnable.addMsg(packet);
        channelContext.handleQueueRunnable.execute();
    }

    /**
     * 清空处理的队列消息
     */
    @Override
    public void clearMsgQueue() {
        super.clearMsgQueue();
        lastByteBuffer = null;
        newReceivedByteBuffer = null;
    }

    @Override
    public ConcurrentLinkedQueue<ByteBuffer> getQueue() {
        return null;
    }

    @Override
    public void runTask() {
        while ((newReceivedByteBuffer = queue.poll()) != null) {
            decode();
        }
    }

    public void decode() {
        ByteBuffer byteBuffer = newReceivedByteBuffer;
        if (lastByteBuffer != null) {
            byteBuffer = ByteBufferUtils.composite(lastByteBuffer, byteBuffer);
            lastByteBuffer = null;
        }

        label_2:
        while (true) {
            try {
                int initPosition = byteBuffer.position();
                int initLimit = byteBuffer.limit();
                int readableLength = initLimit - initPosition;
                Packet packet = null;
                if (channelContext.packetNeededLength != null) {
                    if (log.isInfoEnabled()) {
                        log.info("{}, 解码所需长度:{}", channelContext, channelContext.packetNeededLength);
                    }
                    if (readableLength >= channelContext.packetNeededLength) {
                        packet = config.getTioHandler().decode(byteBuffer, initLimit, initPosition, readableLength, channelContext);
                    }
                } else {
                    try {
                        packet = config.getTioHandler().decode(byteBuffer, initLimit, initPosition, readableLength, channelContext);
                    } catch (BufferUnderflowException e) {
                        //数据不够读
                    }
                }
                // 数据不够，解不了码
                if (packet == null) {
                    byteBuffer.position(initPosition);
                    byteBuffer.limit(initLimit);
                    lastByteBuffer = byteBuffer;
                    //
                    ChannelStat channelStat = channelContext.channelStat;
                    channelStat.decodeFailCount++;

                    if (log.isInfoEnabled()) {
                        log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", channelContext, channelStat.decodeFailCount, readableLength);
                    }
                    if (channelStat.decodeFailCount > 5) {
                        if (channelContext.packetNeededLength == null) {
                            if (log.isInfoEnabled()) {
                                log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", channelContext, channelStat.decodeFailCount, readableLength);
                            }
                        }

                        //检查慢包攻击
                        if (channelStat.decodeFailCount > 10) {

                            int per = readableLength / channelStat.decodeFailCount;
                            if (per < Math.min(channelContext.getReadBufferSize() / 2, 256)) {
                                String str = "连续解码" + channelStat.decodeFailCount + "次都不成功，并且平均每次接收到的数据为" + per + "字节，有慢攻击的嫌疑";
                                throw new DecodeException(str);
                            }
                        }
                    }
                    return;
                } else //解码成功
                {
                    channelContext.setPacketNeededLength(null);
                    channelContext.channelStat.latestTimeOfReceivedPacket = System.currentTimeMillis();
                    channelContext.channelStat.decodeFailCount = 0;

                    int packetSize = byteBuffer.position() - initPosition;
                    packet.setByteCount(packetSize);
                    if (log.isDebugEnabled()) {
                        log.debug("{}, 解包获得一个packet:{}", channelContext, packet);
                    }
                    handle(packet, packetSize);
                    //组包后，还剩有数据
                    if (byteBuffer.hasRemaining()) {
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，还剩有数据:{}", channelContext, byteBuffer.remaining());
                        }
                        continue label_2;
                    } else {//组包后，数据刚好用完
                        lastByteBuffer = null;
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，数据刚好用完", channelContext);
                        }
                        return;
                    }
                }
            } catch (Throwable e) {
                if (channelContext.logWhenDecodeError) {
                    log.error("解码时遇到异常", e);
                }
                channelContext.setPacketNeededLength(null);
                Io.close(channelContext, e, "解码异常:" + e.getMessage(), CloseCode.DECODE_ERROR);
                return;
            }
        }
    }


    private final ConcurrentLinkedQueue<ByteBuffer> queue = new ConcurrentLinkedQueue();

}
